package io.kimmking.dubbo.demo.second.consumer.listeners;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import io.kimmking.common.rocketmq.base.BaseIConsumer;
import io.kimmking.common.rocketmq.base.RocketMQMsgListener;
import io.kimmking.dubbo.common.entity.Order;
import io.kimmking.dubbo.common.mapper.CustomerMapper;
import io.kimmking.dubbo.common.unit.ThrowableActionV2;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.SelectorType;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.nio.charset.Charset;
import java.util.List;

@Slf4j
@Service
@RocketMQMsgListener(
        //下边可以去掉，都使用的是默认值
        consumeMode = ConsumeMode.CONCURRENTLY, //默认值，并发接收模式
        messageModel = MessageModel.CLUSTERING, //默认值，集群模式
        selectorType = SelectorType.TAG, // 默认值，标签
        selectorExpression = "*" //默认值，匹配该topic下所有tag
)
public class ConsumeCanalOrder extends BaseIConsumer<Order> {

    @Resource
    CustomerMapper customerMapper;

    @Value("{kimmking.canal.order.topic:topic_canal_order}")
    private String topic;

    @Value("{kimmking.canal.order.group:group_canal_order}")
    private String cosumer_group;

    @Override
    protected void handle(List<MessageExt> msgs) {
    }

    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {

        msgs.stream()
            .filter(messageExt -> isVerify(messageExt, Order.class))
            .forEach(messageExt -> ThrowableActionV2.execute(()->{
                JSONObject body = JSONArray.parseObject(new String(messageExt.getBody(), Charset.defaultCharset()));
                Integer costomerId = body.getInteger("customerId");
                Float amount = body.getFloatValue("orderCost") * body.getFloatValue("proNum");
                //根据cId更新账号值，todo...
                customerMapper.updateAccountById(amount,costomerId);
            }));

        return null;
    }
}
